package cn.itcast.flink.sql;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

public class OrderUnionDemo {
    public static void main(String[] args) throws Exception {
        //1.准备环境 创建流环境 和 流表环境，并行度设置为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //表环境创建
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //2.Source 创建数据集
        DataStream<Order> orderA = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(1L, "diaper", 4),
                new Order(3L, "rubber", 2)));

        DataStream<Order> orderB = env.fromCollection(Arrays.asList(
                new Order(2L, "pen", 3),
                new Order(2L, "rubber", 3),
                new Order(4L, "beer", 1)));
        //3.注册表 将数据流转换成表
        // 通过fromDataStream将数据流转换成 Table
        Table table = tEnv.fromDataStream(orderA);
        // 将数据流转换成 创建临时视图 View
        tEnv.createTemporaryView("t_orderB",orderB,$("user"),$("product"),$("amount"));
        //4.执行查询，查询order1的amount>2并union all 上 order2的amoun<2的数据生成表
        Table result1 = table
                .where($("amount")
                .isGreater(2))
                .select($("user"),$("product"),$("amount"));

        String explain = tEnv.explainSql("select user,product,amount " +
                "from t_orderB " +
                "where amount<2 " +
                " union all " +
                "select user,product,amount from " + result1);
        System.out.println(explain);

        Table result = tEnv.sqlQuery("select user,product,amount " +
                "from t_orderB " +
                "where amount<2 " +
                " union all " +
                "select user,product,amount from " + result1);
        //4.1 将结果表转换成toAppendStream数据流
        DataStream<Row> rowDataStream = tEnv.toAppendStream(result, Row.class);
        //5.打印结果
        rowDataStream.print();
        //6.执行环境
        env.execute();
        // 创建实体类 user:Long product:String amount:int
    }

    @AllArgsConstructor
    @NoArgsConstructor
    @Data
    public static class Order{
        private Long user;
        private String product;
        private int amount;
    }
}